package com.clw.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.clw.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 识别新老用户
 */
public class NewOrOldUser {
    public static void main(String[] args) throws Exception {
        // 1. 创建流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取kafka 的消费者的流
        FlinkKafkaConsumer<String> logSource = MyKafkaUtil.getKafkaConsumer("dwd_common_log", "safsfsd");


        // 获取流数据
        DataStreamSource<String> commonStream = env.addSource(logSource);

        //识别新老用户
        SingleOutputStreamOperator<JSONObject> isNewStream = commonStream.map(x -> {
            return JSON.parseObject(x);
        }).keyBy(x -> {
            return x.getJSONObject("common").getString("mid");
        }).map(new RichMapFunction<JSONObject, JSONObject>() {
            // 记录上次访问的日期
            ValueState<String> lastvisitdate;
            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");

            @Override
            public void open(Configuration parameters) throws Exception {


                super.open(parameters);
                lastvisitdate = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastvisitdate", String.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                // 只要访问过 就是旧的  没有访问过 就是新的
                String ts = dateFormat.format(new Date(value.getLong("ts")));
                String oldTs = lastvisitdate.value();
                if (oldTs == null || !oldTs.equals(ts)) {
                    value.getJSONObject("common").put("is_new", "1");
                } else {
                    value.getJSONObject("common").put("is_new", "0");
                }
                return value;


            }
        });

        // 发送到kafka
        isNewStream.print();
        env.execute();
    }
}
